package websocket

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/text/gstr"
	"github.com/gogf/gf/v2/util/gconv"

	"github.com/beherochuling/beesoft_api/app/helper/net"
	"github.com/beherochuling/beesoft_api/app/helper/software/software"
)

var cluster = &clusterService{
	ctx:                context.Background(),
	Enabled:            g.Config().MustGet(ctx, "app.websocket.cluster.enabled", true).Bool(),
	serviceTimeout:     g.Config().MustGet(ctx, "app.websocket.serviceTimeout", 3).Uint(),
	redisPubSubChannel: fmt.Sprintf("easywebsocket-cluster-%s", g.Config().MustGet(ctx, "app.name", "")),
}

type clusterService struct {
	ctx                context.Context
	Enabled            bool
	redisPubSubChannel string
	serviceTimeout     uint
	// workersTimeout int64
	workers    map[string]*clusterWorker
	workersMux *sync.Mutex
	listenIp   string
	listenPort int
}
type clusterWorker struct {
	IP            string `json:"ip,omitempty"`
	Port          int    `json:"port,omitempty"`
	LastHeartbeat int64  `json:"-"`
}

func (r *clusterService) Run(s *ghttp.Server) {
	s.Group(uriPrefixFramework, func(group *ghttp.RouterGroup) {
		group.POST("/actions", r.HandleActions)
	})

	r.workers = map[string]*clusterWorker{}
	r.workersMux = &sync.Mutex{}

	if r.serviceTimeout <= 0 {
		r.serviceTimeout = 3
	}

	r.listenIp = net.Net.GetLocalIP()
	address := g.Config().MustGet(r.ctx, "server.address", "").String()
	if address != "" {
		parts := gstr.Split(address, ":")
		if len(parts) == 2 {
			r.listenPort = gconv.Int(parts[1])
		}
	}
	if r.listenPort <= 0 {
		panic(fmt.Sprintf("invalid listen port: '%d'", r.listenPort))
	}

	go func() {
		for {
			err := software.Redis(r.ctx).Subscribe(r.redisPubSubChannel, func(payload string) {
				r.UpdateWorker(payload)
			})
			if err != nil {
				// todo
				// glog.Error(r.ctx, err)
			}

			time.Sleep(time.Second * 3)
		}
	}()
	go func() { // 定时检查当前 Workers 可用性
		for {
			r.CheckWorkers()

			time.Sleep(time.Second * 3)
		}
	}()
	go func() { // 定时广播 IP 端口给其他 Workers
		for {
			r.PublishWorker()
			time.Sleep(time.Second * 3)
		}
	}()

	r.PublishWorker() // 一启动就先立即广播一下自己上线
}

func (r *clusterService) HandleActions(req *ghttp.Request) {
	var action *clusterAction
	err := req.Parse(&action)
	if err != nil {
		net.NewResponse(req).ResponseBadRequest(err.Error())
	}

	switch action.Action {
	case clusterActionSendMessage:
		var msg *clusterSendMessageObject
		err = gconv.Struct(action.Data, &msg)
		if err != nil {
			net.NewResponse(req).ResponseBadRequest(err.Error())
		}

		group.SendMessage(messageFromCluster, msg)
	case clusterActionGetUsers:
		groupId := gconv.String(action.Data)
		localUsers := group.GetLocalUsers(groupId)

		net.NewResponse(req).ResponseOk(localUsers)
	case clusterActionCheckExists:
		var data *clusterCheckExistsObject
		err = gconv.Struct(action.Data, &data)
		if err != nil {
			net.NewResponse(req).ResponseBadRequest(err.Error())
		}

		exist := false
		if data.Type == checkExistTypeGroup {
			exist = group.GroupExists(messageFromCluster, data.GroupId)
		} else if data.Type == checkExistTypeUser {
			exist = group.UserExists(messageFromCluster, data.GroupId, data.UserId)
		}

		net.NewResponse(req).ResponseOk(g.Map{
			"exists": exist,
		})
	case clusterActionCloseConnection:
		var data *clusterCloseConnectionObject
		err = gconv.Struct(action.Data, &data)
		if err != nil {
			net.NewResponse(req).ResponseBadRequest(err.Error())
		}

		group.CloseWebsocket(messageFromCluster, data.GroupId, data.FD)
	default:
		net.NewResponse(req).ResponseBadRequest("invalid action")
	}
}

func (r *clusterService) GetWorkers() []string { // 获取 Workers 地址列表
	var addresses []string

	for address := range r.workers {
		addresses = append(addresses, address)
	}

	return addresses
}
func (r *clusterService) IterateWorkers(callback func(address string) (breakIterator bool)) {
	workers := r.GetWorkers()
	if len(workers) > 0 {
		for _, address := range workers {
			if callback(address) {
				return
			}
		}
	}
}
func (r *clusterService) CheckWorkers() { // 清理超时 Workers
	r.workersMux.Lock()
	defer r.workersMux.Unlock()

	for address, worker := range r.workers {
		if gtime.Now().Timestamp()-worker.LastHeartbeat >= 6 {
			delete(r.workers, address)
		}
	}
}
func (r *clusterService) PublishWorker() {
	_, err := software.Redis(r.ctx).Publish(r.redisPubSubChannel, &clusterWorker{
		IP:   r.listenIp,
		Port: r.listenPort,
	})
	if err != nil {
		// todo
		// glog.Error(r.ctx, err)
	}
}
func (r *clusterService) UpdateWorker(payload string) {
	var worker *clusterWorker
	err := gconv.Struct(payload, &worker)
	if err != nil {
		glog.Error(ctx, err)
		return
	}

	// 排除自己
	if worker.IP == r.listenIp && worker.Port == r.listenPort {
		return
	}

	r.workersMux.Lock()
	defer r.workersMux.Unlock()

	workerAddress := fmt.Sprintf("%s:%d", worker.IP, worker.Port)
	if w, ok := r.workers[workerAddress]; ok {
		w.LastHeartbeat = gtime.Now().Timestamp()
	} else {
		worker.LastHeartbeat = gtime.Now().Timestamp()
		r.workers[workerAddress] = worker
	}
}
